//package com.activemqwrap.tools.tcpforward;
//
//import java.io.InputStream;
//import java.io.OutputStream;
//import java.net.InetAddress;
//import java.net.ServerSocket;
//import java.net.Socket;
//import java.util.Arrays;
//import java.util.Base64;
//
//public class SoketAndMQ {
//    private  final int MAX_CONNECTION_NUM = 50000;
//    private boolean rcvRun = false;
//    private ServerSocket rcvServerSocket = null;
//    public  String[] args = null;
//    public  int sleepTime = 10;
//    public  int buffsize = 1024;
//
////    https://123.58.180.8/ https://123.58.180.7/
////    /**
////     * 读入流的数据的套接字。
////     */
////    private Socket readSocket;
////    /**
////     * 输出数据的套接字。
////     */
////    private Socket writeSocket;
//
//    public void run(int InOrOUt) {
//        try {
//            //1 数据输入
//            if (InOrOUt == 1 || InOrOUt == 0) {
//                String inIp = args[0];
//                int inPort = Integer.parseInt(args[1]);
//                rcvRun = true;
//                rcvServerSocket = new ServerSocket(inPort, MAX_CONNECTION_NUM, InetAddress.getByName(inIp));
//                System.out.println("开始监听：" + inIp + ":" + inPort);
//                new Thread(() -> {
//                    while (rcvRun) {
//                        try {
//                            // 用户连接到当前服务器的socket
//                            Socket rcvSocket = rcvServerSocket.accept();
//                            System.out.println("监听到新接入连接：" + rcvSocket.getRemoteSocketAddress());
//                            new Thread(() -> {
//                                rcvThread(rcvSocket);
//                            }).start();
////                            Thread.sleep(sleepTime);
//
//                        } catch (Exception e) {
//
//                        }
//                    }
//                }).start();
//
//            }
//            //2 数据转发
//            if (InOrOUt == 2 || InOrOUt == 0) {
//                new Thread(() -> {
//                    try {
//                        PushPopActivMQ.rcvMQ_waitingSocket("waitingSocket");
//                        System.out.println("开始监听输入队列：" + "waitingSocket");
//
//                    } catch (Exception e) {
//
//                    }
//                }).start();
//
//            }
//
//        } catch (Exception e) {
//
//        }
//
//    }
//
//    public void stop() {
//        try {
//            rcvRun = false;
//            rcvServerSocket.close();
//        } catch (Exception e) {
//
//        }
//
//    }
//
//    // 接收数据 创建等待队列 插入消息队列
//    public  void rcvThread(Socket rcvSocket) {
//        try {
//            String socketInfo = rcvSocket.getRemoteSocketAddress().toString();
//            PushPopActivMQ.sendMQ("waitingSocket", socketInfo);
//
//            //队列数据 返回输入socket
//            OutputStream outStream = rcvSocket.getOutputStream();
//            returnQueueToSocketThread(socketInfo + "@retQue", outStream);
//
//            byte[] buff = new byte[buffsize];
//            InputStream inStream = rcvSocket.getInputStream();
//            while (!rcvSocket.isClosed()) {
//                int size = inStream.read(buff);
//                if (size > -1) {
//                    byte[] buff_s;
//                    if (size < 1024) {
//                        buff_s = Arrays.copyOf(buff,size);
//                    }else{
//                        buff_s = buff;
//                    }
//                    String Str_Base64 = Base64.getEncoder().encodeToString(buff_s);
////                    byte[] bt_Base64 = Base64.getDecoder().decode(Str_Base64);
//                    PushPopActivMQ.sendMQ(socketInfo + "@rcvQue", Str_Base64);
//                    System.out.println("输入数据转入队列：" + socketInfo);
//                }
//
//                Thread.sleep(sleepTime);
//            }
//        } catch (Exception e) {
//
//        }
//    }
//
//    // 处理接收线程 队列数据返回输入socket
//    public  void returnQueueToSocketThread(String queueName, OutputStream outStream) {
//        new Thread(() -> {
//            try {
//                PushPopActivMQ.MQ_trsToTarget(queueName, outStream);
//            } catch (Exception e) {
//
//            }
//
//        }).start();
//    }
//
//
//    //创建消息队列转发目标socket
//    public  void trsThread(String waiting_msg) {
//        try {
//            // 当前服务器连接到目的地服务器的socket。
//            String outIp = args[2];
//            int outPort = Integer.parseInt(args[3]);
//            Socket toSocket = new Socket(outIp, outPort);
//            if (!toSocket.isClosed()) {
//                //socket监听目标返回
//                returnSocketToQueThread(toSocket, waiting_msg + "@retQue");
//                //接收数据队列转目标socket
//                OutputStream os = toSocket.getOutputStream();
//                PushPopActivMQ.MQ_trsToTarget(waiting_msg + "@rcvQue", os);
//                System.out.println("输入队列数据转socket：" + waiting_msg + "@rcvQue");
//            }
//
//        } catch (Exception e) {
//            String aaa = e.getMessage();
//        }
//
//    }
//
//    //目标返回数据入队列
//    public  void returnSocketToQueThread(Socket toSocket, String queueName) {
//        byte[] buff = new byte[buffsize];
//        new Thread(() -> {
//            try {
//                InputStream rs = toSocket.getInputStream();
//                while (!toSocket.isClosed()) {
//                    int size = rs.read(buff);
//                    if (size > -1) {
//                        byte[] buff_s;
//                        if (size < 1024) {
//                            buff_s = Arrays.copyOf(buff,size);
//                        }else{
//                            buff_s = buff;
//                        }
//                        String retMsg = Base64.getEncoder().encodeToString(buff_s);
//                        PushPopActivMQ.sendMQ(queueName, retMsg);
//                        System.out.println("返回数据转入队列：" + queueName);
//                    }
//                    Thread.sleep(sleepTime);
//                }
//            } catch (Exception e) {
//
//            }
//
//        }).start();
//
//    }
//
////    //向目标ip转发数据
////    public  void trsData(String msg, OutputStream os) {
////        try {
////            byte[] nb = Base64.getDecoder().decode(msg);
////            os.write(nb, 0, nb.length);
////
////        } catch (Exception e) {
////
////        }
////
////    }
//
//
////    public void ReadWriteRunnable(Socket readSocket, Socket writeSocket) {
////        this.readSocket = readSocket;
////        this.writeSocket = writeSocket;
////    }
////
////
////    public void run() {
////        byte[] b = new byte[buffsize];
////        InputStream is = null;
////        OutputStream os = null;
////        try {
////            is = readSocket.getInputStream();
////            os = writeSocket.getOutputStream();
////            while (!readSocket.isClosed() && !writeSocket.isClosed()) {
////                int size = is.read(b);
////                if (size > -1) {
////                    String s = Base64.getEncoder().encodeToString(b);
////                    byte[] nb = Base64.getDecoder().decode(s);
////                    os.write(nb, 0, size);
////                }
////            }
////        } catch (IOException e) {
////            e.printStackTrace();
////        } finally {
////            try {
////                if (is != null) {
////                    is.close();
////                }
////                if (null != os) {
////                    os.flush();
////                    os.close();
////                }
////            } catch (IOException e) {
////                e.printStackTrace();
////            }
////        }
////
////    }
//
//
////    public void socketToMQ(Socket sk) {
////        byte[] b = new byte[buffsize];
////        InputStream is = null;
////        OutputStream os = null;
////        try {
////            is = sk.getInputStream();
////            while (!sk.isClosed()) {
////                int size = is.read(b);
////                if (size > -1) {
////                    String strMQ = Base64.getEncoder().encodeToString(b);
//////                    byte[] nb = Base64.getDecoder().decode(s);
//////                    PushPopActivMQ.sendMQ(strMQ);
////
////                }
////            }
////
////        } catch (Exception e) {
////            e.printStackTrace();
////        } finally {
////
////        }
////    }
//
//
//}
